/*
    消费者(订阅者)相关类的定义: 消费者类、消费者队列类、消费者管理类
    1. 消费者类：对消费者的描述，只有订阅消息队列的时候才会产生消费者对象
        属性：标签、队列名称、自动应答标志、回调函数
    注：标签：消费者的唯一标识
        队列名：消费者订阅的是哪个队列的消息
        自动应答标志：是否需要等待此消费者的应答（当消息推送给这个消费者的时候，是否需要等待应答，取决于这个消费者），true表示不需要应答，false表示需要
        回调函数：此消费者处理消息的方法（当队列发布一个消息的时候，就需要选择一个消费者进行消费（***待完善***））
    2. 消费者队列类：消费者以队列为单位进行管理
        属性：消费者列表、轮询序号
        方法：新增消费者、删除消费者、选择一个消费者
    注：以队列为单位进行管理的思想和消息持久化的管理思想一致
    3. 消费者管理类：对消费者队列进行管理
        属性：消费者队列映射表<队列名, 消费者队列对象>
        方法：向指定队列新增消费者、从指定队列删除消费者、从指定队列选择一个消费者、销毁指定消费者队列
*/

#pragma once
#include "../common/helper.hpp"
#include "../common/logger.hpp"
#include "../common/message.pb.h"
#include <vector>
#include <memory>
#include <functional>
#include <mutex>

namespace mq
{
    // 消费者类
    struct Consumer
    {
        using ConsumerPtr = std::shared_ptr<Consumer>;
        using Callback = std::function<void(const std::string& tag, const BasicProperties *bp, const std::string& body)>;

        Consumer() {}
        Consumer(const std::string &tag, const std::string &queue_name, bool auto_ack, const Callback &callback)
            : _tag(tag), _queue_name(queue_name), _auto_ack(auto_ack), _callback(callback)
        {
        }

        std::string _tag;        // 标签
        std::string _queue_name; // 队列名称
        bool _auto_ack;          // 自动应答标志
        Callback _callback;      // 回调函数
    };

    // 消费者队列管理类
    class ConsumerQueue
    {
    public:
        using ConsumerQueuePtr = std::shared_ptr<ConsumerQueue>;

    public:
        ConsumerQueue() : _index(0)
        {
        }

        // 新增消费者
        Consumer::ConsumerPtr increaseConsumer(const std::string &tag, const std::string &queue_name, bool auto_ack, const Consumer::Callback &callback)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto &consumer : _consumers)
            {
                if (consumer->_tag == tag)
                    return {};
            }
            Consumer::ConsumerPtr consumer_ptr = std::make_shared<Consumer>(tag, queue_name, auto_ack, callback);
            _consumers.push_back(consumer_ptr);
            return consumer_ptr;
        }

        // 删除消费者
        void removeConsumer(const std::string &tag)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto it = _consumers.begin(); it != _consumers.end(); ++it)
            {
                if ((*it)->_tag == tag)
                {
                    _consumers.erase(it);
                    return;
                }
            }
        }

        // 轮询选择一个消费者
        Consumer::ConsumerPtr choose()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (_consumers.size() == 0)
                return {};
            size_t tmp = _index;
            _index = (_index + 1) % _consumers.size();
            return _consumers[tmp];
        }

        /*
            下面是和测试相关的代码
        */

        // 判断消费者是否存在
        bool hasConsumer(const std::string &tag)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto &consumer : _consumers)
            {
                if (consumer->_tag == tag)
                    return true;
            }
            return false;
        }

        // 判断消费者是否为空
        bool empty()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _consumers.size() == 0;
        }

    private:
        std::vector<Consumer::ConsumerPtr> _consumers;
        size_t _index;
        std::mutex _mutex;
    };

    // 消费者管理类
    class ConsumerManager
    {
    public:
        using ConsumerManagerPtr = std::shared_ptr<ConsumerManager>;

    public:
        ConsumerManager() {}

        // 向指定队列新增消费者
        Consumer::ConsumerPtr increaseConsumer(const std::string &tag, const std::string &queue_name, bool auto_ack, const Consumer::Callback &callback)
        {
            ConsumerQueue::ConsumerQueuePtr consumer_queue_ptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _consumer_queues.find(queue_name);
                if(it == _consumer_queues.end())
                {
                    consumer_queue_ptr = std::make_shared<ConsumerQueue>();
                    _consumer_queues.emplace(queue_name, consumer_queue_ptr);
                }
                else
                {
                    consumer_queue_ptr = it->second;
                }
            }
            return consumer_queue_ptr->increaseConsumer(tag, queue_name, auto_ack, callback);
        }

        // 删除指定队列的消费者
        void removeConsumer(const std::string& queue_name, const std::string &tag)
        {
            ConsumerQueue::ConsumerQueuePtr consumer_queue_ptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _consumer_queues.find(queue_name);
                if(it == _consumer_queues.end())
                {
                    DBG_LOG("queue_name: %s not exist", queue_name.c_str());
                    return;
                }
                consumer_queue_ptr = it->second;
            }
            consumer_queue_ptr->removeConsumer(tag);
        }

        // 轮询选择指定队列的一个消费者
        Consumer::ConsumerPtr choose(const std::string &queue_name)
        {
            ConsumerQueue::ConsumerQueuePtr consumer_queue_ptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _consumer_queues.find(queue_name);
                if(it == _consumer_queues.end())
                {
                    return {};
                }
                consumer_queue_ptr = it->second;
            }
            return consumer_queue_ptr->choose();
        }

        // 销毁指定消费者队列
        void destroyConsumerQueue(const std::string &queue_name)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _consumer_queues.erase(queue_name);
        }

        /*
            下面是和测试相关的代码
        */
        // 指定队列的消费者是否存在
        bool hasConsumer(const std::string &queue_name, const std::string& tag)
        {
            ConsumerQueue::ConsumerQueuePtr consumer_queue_ptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _consumer_queues.find(queue_name);
                if(it == _consumer_queues.end())
                {
                    DBG_LOG("queue_name: %s not exist", queue_name.c_str());
                    return false;
                }
                consumer_queue_ptr = it->second;
            }
            return consumer_queue_ptr->hasConsumer(tag);
        }

        // 指定消费者队列是否为空
        bool empty(const std::string &queue_name)
        {
            ConsumerQueue::ConsumerQueuePtr consumer_queue_ptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _consumer_queues.find(queue_name);
                if(it == _consumer_queues.end())
                    return true;
                consumer_queue_ptr = it->second;
            }
            return consumer_queue_ptr->empty();
        }

        // 清空消费者
        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _consumer_queues.clear();
        }

    private:
        std::unordered_map<std::string, ConsumerQueue::ConsumerQueuePtr> _consumer_queues;
        std::mutex _mutex;
    };
}